package org.swift.mashup.engine.concurrency.execution;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.BitSet;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import org.apache.axis2.context.MessageContext;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.swift.mashup.engine.IndexedFile;
import org.swift.mashup.engine.MashupException;
import org.swift.mashup.engine.concurrency.ConcurrentContext;
import org.swift.mashup.engine.concurrency.agenda.*;
import org.swift.mashup.engine.concurrency.memory.MappedMemoryManager;
import org.swift.mashup.engine.concurrency.memory.MemoryEvent;
import org.swift.mashup.engine.concurrency.memory.MemoryEventListener;
import org.swift.mashup.engine.concurrency.memory.MemoryItem;
import org.swift.mashup.engine.concurrency.memory.MemoryType;
import org.swift.mashup.engine.concurrency.memory.WorkingMemory;
import org.swift.mashup.engine.model.ActionOutput;

import edu.emory.mathcs.backport.java.util.Collections;

@SuppressWarnings({"unchecked","rawtypes"})
public class Reasoner implements MemoryEventListener, AgendaListener {
	private static Log log = LogFactory.getLog(Reasoner.class);
	private RuleSet ruleSet = null;
	private WorkingMemory memory = null;
	private boolean parallel;
	private boolean laxErrors;
	private Integer agendaTimeout;
    private Integer rateLimit = null;
	private ConcurrentContext concurrentContext;
	private List<AgendaItem> localAgendaItems = new ArrayList<AgendaItem>();
	
	private int indexFileCount = 0;
	
	public Set<RulePattern> patterns;

	public static String ERROR_TYPE = "ERROR";

	public Reasoner(RuleSet ruleSet) {
		this.ruleSet = ruleSet;
		this.memory = new WorkingMemory(ruleSet.getMemoryTypes());
	}

	public Reasoner(RuleSet ruleSet, WorkingMemory memory,
			ConcurrentContext concurrentContext) {
		this.ruleSet = ruleSet;
		this.memory = memory;
		this.parallel = (concurrentContext != null && concurrentContext
				.getComputeNodeCount() > 1);
		this.concurrentContext = concurrentContext;
	}

	protected boolean isParallel() {
		return parallel;
	}

	protected void setParallel(boolean parallel) {
		this.parallel = parallel;
	}

	protected boolean isLaxErrors() {
		return laxErrors;
	}

	protected void setLaxErrors(boolean laxErrors) {
		this.laxErrors = laxErrors;
	}

	protected Integer getAgendaTimeout() {
		return agendaTimeout;
	}

	protected void setAgendaTimeout(int timeout) {
		this.agendaTimeout = timeout;
	}

	public Map<String, Rule> getRules() {
		return ruleSet.getRules();
	}

	public Set<RulePattern> getPatterns() {
		return patterns;
	}

	public void setPatterns(Set<RulePattern> patterns) {
		this.patterns = patterns;
	}

	public ExecutionOutput fire(Map<String, ?> items) {
		ExecutionOutput rs;
		try {
			memory.addListener(this);

			List<RulePattern> activeRulePatterns = new ArrayList<RulePattern>();
			activeRulePatterns.addAll(ruleSet.matchPatterns(new BitSet(),
					new BitSet()));
			log.debug("Evaluating initial rule set");
			Collection<AgendaItem> agendaItems = evaluateRules(activeRulePatterns,
					null, this.memory);
			log.debug("Executing initial agenda");
			this.runAgendaItems(agendaItems);

			Map<String, Object> memoryMap = new HashMap(ruleSet.getDefaultValues());
			for (Map.Entry<String,?> entry : items.entrySet() ) {
				if ( ruleSet.getMemoryTypes().containsKey(entry.getKey()) ) {
					memoryMap.put(entry.getKey(), entry.getValue());
				}
			}

			this.memory.insertAll(memoryMap);
			int generation = 0;
			while (this.memory.hasNewMemoryItems()) {
				this.memory.flush(this.parallel);
				if (this.parallel && this.concurrentContext.getJobQueueSize() > 0) {
					this.concurrentContext.runJobs();
					collectRemoteMemoryItems();
				} 
				log.debug("Executing agenda for generation " + (++generation));
				if (this.localAgendaItems.size() > 0) {
					this.runAgendaItems(this.localAgendaItems);
					this.localAgendaItems.clear();				
				}
				if (!this.memory.hasNewMemoryItems()) {
					this.memory.addCollectsToBuffer(this.parallel);
				}
			}
			Map<String, Collection<Object>> map = new HashMap<String, Collection<Object>>();
			for (MemoryType type : ruleSet.getMemoryTypes().values()) {
				if (type.isPublish() && memory.getItems(type.getName()).size() > 0) {
					List<Object> list = new ArrayList<Object>();
					map.put(type.getName(), list);
					for (MemoryItem item : memory.getItems(type.getName())) {
						list.add(item.toString());
					}
				}
			}
			Set<ActionException> exceptions = this.getErrors();
			if (exceptions.size() > 0){
				rs = new ExecutionOutput(map,exceptions);
			} else {
				rs = new ExecutionOutput(map);
			}
		}
		finally {
			if (this.parallel) {
				if (this.concurrentContext.getConfiguration().isDeleteFiles()){
					this.memory.destroyMemoryFiles();
					this.memory.deleteMemoryFiles();
				}
			}
			else {
				this.memory.clear();
			}
		}
		return rs;
	}
	
	public void runAgendaItems(Collection<AgendaItem> agendaItems) {
        AgendaConfig cfg = new AgendaConfig();
        if ( rateLimit != null ) {
            cfg.setRateLimit(rateLimit);
        }
        if ( agendaTimeout != null ) {
            cfg.setAgendaTimeout(agendaTimeout);
        }

		ConcurrentAgenda agenda = new ConcurrentAgenda(agendaItems, cfg);
		agenda.setLaxErrors(laxErrors);
		agenda.addListener(this);
		if(AgendaLogger.isEnabled()){
			agenda.addListener(new AgendaLogger());
		}
		agenda.run();
	}

	private void collectRemoteMemoryItems() {
		Set<MemoryType> newTypes = new HashSet<MemoryType>();
		ConcurrentContext context = this.concurrentContext;
		String memoryDir = context.getWorkingDirectory() + "/workingMemory";
		File outputDir = new File(memoryDir);
		for (int i = context.getProcessedJobs(); i < context.getTotalJobs(); i++) {
			int jobNumber = i + 1;
			String sessionId = this.concurrentContext.getSessionID().toString();
			String filename = sessionId + "-" + jobNumber;
			File inputMapFile = new File(outputDir, "mm-" + filename + ".map");
			if (!inputMapFile.exists())
				break;
			MappedMemoryManager.loadMemory(memory, sessionId, "" + jobNumber,
					-1, -1, true);
			newTypes.addAll(this.memory.getBufferedMemoryTypes());
			context.setProcessedJobs(jobNumber);
		}
		Map<String, MemoryType> memoryTypes = this.ruleSet.getMemoryTypes();
		for (MemoryType memType : newTypes) {
			if (memType.isCollect()) {
				MemoryType collectType = memoryTypes.get("collect("
						+ memType.getName() + ")");
				this.memory.insert(collectType.getName(), null);
			}
		}

	}

	public void fireConcurrent() {
		memory.addListener(this);
		this.memory.flush(this.parallel);
		this.runAgendaItems(this.localAgendaItems);
	}

	public void handleEvent(MemoryEvent event) {

		switch (event.getType()) {
		case AddedToWorkingMemory:
			handleAddToWorkingMemoryEvent(event);
		case MemoryFlushed:
			if (this.parallel) {
				int nodeCount = this.concurrentContext.getComputeNodeCount();
				double remainingItems = event.getEndId() - event.getStartId() + 1;
				if (remainingItems == 1) {
					// process the data local because there is only one item
					handleAddToWorkingMemoryEvent(event);
				} else {
					// Process the items remotely
					try {
						this.memory.getWriteableMemory().getMemoryFileChannel()
								.force(true);
					} catch (IOException e) {
						e.printStackTrace();
					}
					double groupSize = Math.ceil(remainingItems / nodeCount);
					int startIndex = event.getStartId();
					while (remainingItems > 0) {
						try {
							remainingItems = remainingItems - groupSize;
							this.concurrentContext.createJobSet(startIndex,
								nodeCount);
							startIndex++;
						} catch (MashupException me) {
							throw new RuntimeException ("Failed to create the job set.",me);
						}
						
					}
				}
			}
		}
	}

	private void handleAddToWorkingMemoryEvent(final MemoryEvent event) {
		final WorkingMemory memory = event.getWorkingMemory();
		final MemoryItem item = event.getItem();
		final BitSet itemMask = item.getType().getMask();
		final BitSet memoryMask = new BitSet();
		memoryMask.or(memory.getWorkingMemoryMask());
		final List<RulePattern> activeRulePatterns = new ArrayList<RulePattern>();
		// get all of the rules that might match for this item type
		final List<RulePattern> matching = ruleSet.matchPatterns(itemMask, memoryMask);
		if(log.isDebugEnabled()) {
			log.debug("Found " + matching.size() + " active rules for item type "
					+ item.getType().getName());
		}
		activeRulePatterns.addAll(matching);
		// now evaluate them against this item & working memory, adding any
		// actions in matching rules to the agenda
		this.localAgendaItems.addAll(evaluateRules(activeRulePatterns, item,
				memory));
	}

	public void handleEvent(AgendaEvent event) {
		AgendaItem item = event.getItem();
		Action action = null;
		switch (event.getType()) {
		case AgendaActionItemCompleted:
			action = ((Action) item.getItem());
			ActionOutput actionOutput = action.getActionResult();
			Map<String,Collection> resultMap = new HashMap<String,Collection>();
			actionOutput.updateOutputMap(resultMap, item.getReturnType());
			for (Entry<String,Collection> entry : resultMap.entrySet()){
				String type = entry.getKey();
				Collection outputs = entry.getValue();
				for (Object output : outputs){
					if (Collection.class.isAssignableFrom(output.getClass())) {
						for (Object oitem : (Collection) output) {
							this.memory.insert(type, oitem);
						}
					} else if (IndexedFile.class.isAssignableFrom(output.getClass())) {
						final String fileIndex = this.memory.getWriteableMemory().getId() + "." + indexFileCount++;
						MappedMemoryManager.loadMemory(this.memory, this.concurrentContext.getSessionID(), fileIndex, type, (IndexedFile)output);
					} else if (ActionException.class
							.isAssignableFrom(output.getClass()) && !laxErrors) {
						this.memory.insert(ERROR_TYPE, output);
					} else {
						this.memory.insert(type, output);
					}
				}
			}
			break;
		case AgendaItemTimedout:
			if(!laxErrors) {
				action = (Action) item.getItem();
				ActionOutput output = action.getActionResult();
				ActionException e = (ActionException) output.getOutput();
				this.memory.insert(ERROR_TYPE, e);
			}
			break;
		}
	}

	private Collection<AgendaItem> evaluateRules(List<RulePattern> patterns,
			MemoryItem item, WorkingMemory memory) {
		List<AgendaItem> agendaItems = new ArrayList<AgendaItem>();
		for (RulePattern pattern : patterns) {
			List<List> parameterSets = buildParameterList(pattern.getTypes(),
					item, memory);
			for (List<MemoryItem> parameters : parameterSets) {
				agendaItems.addAll(evaluatePattern(pattern, parameters));
			}
		}
		return agendaItems;
	}

	private Collection<AgendaItem> evaluatePattern(RulePattern pattern,
			List<MemoryItem> parameterList) {
		Map<String, MemoryItem> parameters = new HashMap<String, MemoryItem>();
		for (MemoryItem item : parameterList) {
			parameters.put(item.getType().getName(), item);
		}
		Collection<Rule> rules = pattern.getActiveRules(parameters);
		if (rules.size() > 0)
			log.debug("\nPattern [" + pattern.getId() + "] matched for Rules ["
					+ rules + "]");
		List<AgendaItem> agendaItems = new ArrayList<AgendaItem>();
		for (Rule rule : rules) {
			agendaItems.addAll(createAgendaItems(rule, parameters));
		}
		return agendaItems;
	}

	private Collection<AgendaItem> createAgendaItems(Rule rule,
			Map<String, MemoryItem> parameters) {
		final List<ActionTemplate> actions = rule.getActions();
		final String returnType = rule.getName();
		final List agendaItems = new ArrayList<AgendaItem>();
		if(log.isDebugEnabled()) {
			final int numActions = actions == null ? 0 : actions.size();
			log.debug("Creating " + numActions + " agenda items for rule "
					+ returnType);
		}
		for (ActionTemplate template : actions) {
			Action action = template.newAction(parameters);
			AgendaItem item = new AgendaItem(action, returnType);
			item.setContext(MessageContext.getCurrentMessageContext());
			agendaItems.add(item);
		}
		return agendaItems;
	}

	private List buildParameterList(Map<String, MemoryType> types,
			MemoryItem item, WorkingMemory memory) {
		final List parameterSets = new ArrayList();
		final List<List> itemLists = new ArrayList<List>();
		for (MemoryType type : types.values()) {
			final String typeName = type.getName();
			final String filterTypeName = item.getType().getName();
			if (item == null || !(typeName.equals(filterTypeName))) {
				log.debug("Fetching items of type " + typeName
						+ " from working memory to use as "
						+ " rule set parameters");
				itemLists.add(memory.getItems(typeName));
			}
		}
		buildParameterSets(parameterSets, new ArrayList(), item, itemLists);
		return parameterSets;
	}

	private void buildParameterSets(List<List> paramSets,
			List<MemoryItem> itemStack, MemoryItem fixedItem,
			List<List> itemLists) {
		if (fixedItem != null) {
			if(log.isDebugEnabled()) {
				log.debug("Adding item type " + fixedItem.getType().getName() + ":"
					+ fixedItem.getType().getId() + " to item stack");
			}
			itemStack.add(fixedItem);
		}
		if (itemLists.size() > 1) {
			List<MemoryItem> itemList = itemLists.get(0);
			List subList = createSubList(itemLists);
			for (MemoryItem item : itemList) {
				if(log.isDebugEnabled()) {
					MemoryType type = item.getType();
					log.debug("Making recursive call to get buildParameterSets() "
						+ "for item type " + type.getName() + ":"
						+ type.getId());
				}
				buildParameterSets(paramSets, itemStack, item, subList);
			}
		} else if (itemLists.size() == 1) {
			List<MemoryItem> itemList = itemLists.get(0);
			for (MemoryItem item : itemList) {
				List<MemoryItem> paramSet = new ArrayList(itemStack);
				if(log.isDebugEnabled()) {
					MemoryType type = item.getType();
					log.debug("Adding item type " + type.getName() + ":"
							+ type.getId() + " to parameter sets");
				}
				paramSet.add(item);
				paramSets.add(paramSet);
			}
		} else {
			log.debug("Adding item stack to parameter sets");
			List<MemoryItem> paramSet = new ArrayList(itemStack);
			paramSets.add(paramSet);
		}
		if (fixedItem != null) {
			if(log.isDebugEnabled()) {
				log.debug("Removing item type " + fixedItem.getType().getName() + ":"
					+ fixedItem.getType().getId() + " from item stack");
			}
			itemStack.remove(fixedItem);
		}
	}

	private List createSubList(List list) {
		List subList = new ArrayList();
		for (int i = 1; i < list.size(); i++) {
			subList.add(list.get(i));
		}
		return subList;
	}
	
	public Set<ActionException> getErrors(){
		Set<ActionException> items = new HashSet<ActionException>();
		for (MemoryItem item : memory.getItems(ERROR_TYPE)) {
			Object val = item.getValue();
			if(val instanceof byte[]){
				String msg = new String((byte[])val);
				ActionException ae = new ActionException("",msg,null);
				items.add(ae);
			}else{
				items.add((ActionException)val);
			}
		}
		return Collections.unmodifiableSet(items);
	}

    public Integer getRateLimit() {
        return rateLimit;
    }

    public void setRateLimit(Integer rateLimit) {
        this.rateLimit = rateLimit;
    }
}
